package com.stress.testing.plugin.apache.rocketmq.interceptor;

import com.stress.testing.common.interceptor.AbstractInterceptor;
import com.stress.testing.plugin.apache.rocketmq.context.StressTestingConsumerHolder;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import java.lang.reflect.Method;

import static com.stress.testing.common.conf.PluginConfigKey.DEFAULT_PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX;
import static com.stress.testing.common.conf.PluginConfigKey.PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX_KEY;

public class DefaultMQPushConsumerUnsubscribeInterceptor extends AbstractInterceptor {
    @Override
    public void before(Method method, Object[] args, Object zuper, Object thiz) {
        if (args == null || args.length == 0) {
            return;
        }

        Object arg = args[0];
        if (arg == null || !(arg instanceof String)) {
            return;
        }

        String topicPrefix = getPluginConfig().get(PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX_KEY,
                DEFAULT_PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX);
        String topic = topicPrefix + arg;


        if (thiz == null || !(thiz instanceof DefaultMQPushConsumer)) {
            return;
        }

        DefaultMQPushConsumer consumer = (DefaultMQPushConsumer) thiz;
        String consumerGroup = consumer.getConsumerGroup();
        if (consumerGroup != null && consumerGroup.startsWith(topicPrefix)) {
            return;
        }

        DefaultMQPushConsumer pushConsumer = StressTestingConsumerHolder.getPushConsumer(consumer.hashCode());
        if (pushConsumer != null) {
            pushConsumer.unsubscribe(topic);
        }
    }
}
